package com.hlz.rabbit.mq.confirm;

import com.hlz.rabbit.mq.util.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 异步确认发布
 *
 * @author Hongliang Zhu
 * @create 2022-11-28 23:02
 */
public class PubModConfirmAsyncProducer {
    public static void main(String[] args) {

        try (Channel channel = RabbitMqUtils.getChannel()) {
            // 开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outStandingConfirm = new ConcurrentSkipListMap<>();
            /*
            确认收到消息的一个回调
            1. 消息序列号
            2. true可以确认小于等于当前序列号的消息
            false表示确认当前序列号

             */
            ConfirmCallback confirmCallback = new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple) {
                        //返回的是小于等于当前序列号的未确认消息，是一个 map
                        ConcurrentNavigableMap<Long, String> confirmed = outStandingConfirm.headMap(deliveryTag, true);
                        // 清除该部分
                        confirmed.clear();
                    } else {
                        // 只清除当前序列号的消息
                        outStandingConfirm.remove(deliveryTag);
                    }
                }
            };

            ConfirmCallback nackCallBack = new ConfirmCallback() {
                @Override
                public void handle(long deliveryTag, boolean multiple) throws IOException {
                    String message = outStandingConfirm.get(deliveryTag);
                    System.out.println("发布的消息" + message + "未被确认， 序列号为：" + deliveryTag);
                }
            };

            // 添加异步确认的监听器
            channel.addConfirmListener(confirmCallback, nackCallBack);


            channel.queueDeclare("queue-pub-confirm", false, false, false, null);
            long start = System.currentTimeMillis();
            for (int i = 0; i < 200; i++) {
                String message = "confirm-async" + i;
                outStandingConfirm.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", "queue-pub-confirm", null, message.getBytes());
            }

            long end = System.currentTimeMillis();

            //异步发布确认的方式：发布 200个消息,耗时15ms ，性能最好！
            System.out.println("异步发布确认的方式：发布 200个消息,耗时" + (end - start) + "ms");

        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
